package rpcx

import (
	"context"
	"github.com/rpcxio/libkv/store"
	"github.com/rpcxio/rpcx-etcd/serverplugin"
	rpcxLog "github.com/smallnest/rpcx/log"
	"github.com/smallnest/rpcx/server"
	"go.uber.org/zap"
	"strings"
	"sync/atomic"
	"time"
	"wzgames/pkg/logger"
	"wzgames/pkg/utils/stringutil"
)

type RPCx struct {
	serv     *server.Server
	conf     *RPCxConfig
	servMeta string
	stopped  int32
}

// RPCx Server
// github.com/smallnest/rpcx for more information

func NewRPCx(conf *RPCxConfig) *RPCx {
	serv := server.NewServer()
	if conf.UpdateInterval == 0 {
		conf.UpdateInterval = 10
	}
	rpcxLog.SetLogger(logger.GetLogger())
	return &RPCx{
		serv:     serv,
		conf:     conf,
		servMeta: genServiceMeta(conf),
	}
}

func (r *RPCx) Start(ctx context.Context) error {
	option := &store.Config{}
	if r.conf.Username != "" {
		option.Username = r.conf.Username
		option.Password = r.conf.Password
	}
	etcdR := &serverplugin.EtcdV3RegisterPlugin{
		ServiceAddress: r.conf.ServAddr,
		EtcdServers:    r.conf.RegServs,
		Options:        option,
		BasePath:       r.conf.BasePath,
		//Metrics:        metrics.NewRegistry(),
		UpdateInterval: time.Duration(r.conf.UpdateInterval) * time.Second,
	}

	err := etcdR.Start()
	if err != nil {
		panic("[RPCx] start Rpcx Server fatal" + err.Error())
	}
	r.serv.Plugins.Add(etcdR)

	// register service to rpcx
	for name, service := range services {
		meta := r.servMeta
		if len(service.meta) > 0 {
			meta += "&" + service.meta
		}
		e := r.serv.RegisterName(name, service.svr, meta)
		if e != nil {
			logger.Error("[RPCx] register service error", zap.String("service", name), zap.Error(e))
		}
	}
	logger.Info("[RPCx] rpcx server start", zap.String("Serv", r.conf.ServAddr))
	network, addr := splitNetworkAndAddress(r.conf.BindAddr)

	go func() {
		err = r.serv.Serve(network, addr)
		if err != nil {
			logger.Error("[RPCx] rpcx server start fail", zap.Error(err))
		}
	}()

	return err
}

func splitNetworkAndAddress(server string) (string, string) {
	ss := strings.SplitN(server, "@", 2)
	if len(ss) == 1 {
		return "tcp", server
	}

	return ss[0], ss[1]
}

func (r *RPCx) Stop(ctx context.Context) error {
	logger.Info("[RPCx] server shutdown", zap.String("Serv", r.conf.ServAddr))
	r.setStopped()

	if r.serv != nil {
		err := r.serv.UnregisterAll()
		if err != nil {
			logger.Error("[RPCx] service unregisterall error", zap.Error(err))
		}
		err = r.serv.Close()
		if err != nil {
			logger.Error("[RPCx] server shutdown error", zap.Error(err))
		}
		return err
	}

	return nil
}

func (r *RPCx) setStopped() {
	atomic.StoreInt32(&r.stopped, 1)
}

func (r *RPCx) isStopped() bool {
	return atomic.LoadInt32(&r.stopped) == 1
}

func genServiceMeta(serv *RPCxConfig) string {
	meta := "wz_svr_id=" + stringutil.ToString(serv.ServID) + "&wz_node=" + stringutil.ToString(serv.NodeId) +
		"&wz_weight=" + stringutil.ToString(serv.Weight) + "&wz_qps=" + stringutil.ToString(serv.MaxQPS)
	return meta
}
